/**
 * Copyright 2009 Erick Armbrust 
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License. 
 * You may obtain a copy of the License at 
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0 
 *   
 * Unless required by applicable law or agreed to in writing, software 
 * distributed under the License is distributed on an "AS IS" BASIS, 
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 * See the License for the specific language governing permissions and 
 * limitations under the License. 
 */

package tapioca.dt;

import java.util.List;
import java.util.Map;

import org.easymock.EasyMock;

import tapioca.dt.WrappedDatastoreService.GetResponse;
import tapioca.testing.AppEngineTestCase;

import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.EntityNotFoundException;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.api.datastore.Query;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.repackaged.com.google.common.collect.Lists;

/**
 * Tests for the {@link DistributedTransaction} class.
 *
 * @author armbrust@gmail.com (Erick Armbrust) 
 */
public class DistributedTransactionTest extends AppEngineTestCase {

  private WrappedDatastoreService wds;
  
  @Override
  public void setUp() throws Exception {
    super.setUp();
    wds = new WrappedDatastoreService(ds);
  }
  
  @Override
  public void tearDown() throws Exception {
    wds = null;
    super.tearDown();
  }  
  
  ///// Helper functions /////
  
  /**
   * Assert that a given {@link CacheRecord} has a particular state.
   */
  private void assertCacheRecordProps(
      CacheRecord record,
      Map<String, Object> properties,
      boolean inserted,
      boolean created,
      boolean retrieved,
      boolean deleted,
      Key version) {
    if (properties != null) {
      assertNotNull(record.getEntity());
      assertEquals(properties, record.getEntity().getProperties());
    }
    assertEquals(inserted, record.isInserted());
    assertEquals(created, record.isCreated());
    assertEquals(retrieved, record.isRetrieved());
    assertEquals(deleted, record.isDeleted());
    if (retrieved) {
      assertEquals(version, record.getVersion());
    }
  }
  
  /**
   * Assert that all the shadow entities created by a given DT are delete from the datastore.
   */
  private void assertShadowsCompletelyRemoved(DistributedTransaction dt) {
    for (PutRequest putRequest : dt.getPuts()) {
      try {
        ds.get(putRequest.getShadowKey());
        fail("DT not completely cleaned up, a shadow copy remains.");
      } catch (EntityNotFoundException e) { /* Good */ }
    }    
  }
  
  /**
   * Assert that a DT is completely cleaned up.  This means that all generated shadow entities 
   * are removed as well as the actual DT entity.
   */
  private void assertDtCompletelyRemoved(DistributedTransaction dt) {
    assertShadowsCompletelyRemoved(dt);
    
    try {
      ds.get(dt.getKey());
      fail("The DT entity should be removed.");
    } catch (EntityNotFoundException e) { /* Good */ }    
  }
  
  ///// Basic CRUD operations /////
  
  public void testCreationAssignsId() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createDtEntity("foo", "bar");
    Entity f2 = createDtEntity("foo", "bar");
    
    dt.put(f1);
    dt.put(f2);
    
    for (CacheRecord record : dt.getCache().values()) {
      assertTrue(record.getEntity().getKey().isComplete());
      assertEquals(f1.getProperties(), record.getEntity().getProperties());
    }    
  }
  
  /**
   * Test that creating an entity with a generated ID within a transaction is properly executed.
   */
  public void testCreateEntity() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createDtEntity("foo", "bar");
    Entity f2 = createDtEntity("fab", "bab");
    
    dt.begin();
    Key k1 = dt.put(f1);
    Key k2 = dt.put(f2);
    dt.commit();

    GetResponse response = wds.get(k1);
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
    assertEquals("bar", response.getEntity().getProperty("value"));
    
    response = wds.get(k2);
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
    assertEquals("bab", response.getEntity().getProperty("value"));    
    
    assertDtCompletelyRemoved(dt);
  }

  /**
   * Test that creating a named entity within a transaction is properly executed.
   */
  public void testCreateNamedEntity() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createDtNamedEntity("foo", "bar", "dt__baz");
    Entity f2 = createDtNamedEntity("fab", "bab", "dt__baz");
    
    dt.begin();
    dt.put(f1);
    dt.put(f2);
    dt.commit();

    GetResponse response = wds.get(f1.getKey());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
    assertEquals("bar", response.getEntity().getProperty("value"));
    
    response = wds.get(f2.getKey());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
    assertEquals("bab", response.getEntity().getProperty("value"));        
    
    assertDtCompletelyRemoved(dt);
  }  
  
  public void testCreateNamedEntity_withVersionCheck() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Key key = KeyFactory.createKey("foo", "dt__bar");
    
    dt.begin();
    try {
      dt.get(key);
    } catch (EntityNotFoundException e) {
      Entity entity = createDtNamedEntity("foo", "baz", "dt__bar");
      dt.put(entity);
    }
    dt.commit();
    
    GetResponse response = wds.get(key);
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
  }  
  
  public void testCreateNamedEntity_failsIfOtherDtInsertsFirst() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Key key = KeyFactory.createKey("foo", "dt__bar");
    
    dt.begin();
    try {
      dt.get(key);
    } catch (EntityNotFoundException e) {
      Entity entity = createDtNamedEntity("foo", "baz", "dt__bar");
      dt.put(entity);
    }
    
    createAndSaveDtNamedEntity("foo", "blah", "dt__bar");
    try {
      dt.commit();
      fail("The version of this entity should have changed.");
    } catch (DistributedTransactionFailedException e) {
      assertTrue(e.getCause() instanceof ReadVersionChangedException);
    }
    
    GetResponse response = wds.get(key);
    assertEquals(dummyVersion, response.getVersion()); 
    assertEquals("blah", response.getEntity().getProperty("value"));
  }  
  
  /**
   * Test that updating an entity within a transaction is properly executed.
   */
  public void testUpdateEntity() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createAndSaveDtEntity("fab", "bab");

    dt.begin();
    Entity f1Dt = dt.get(f1.getKey());
    Entity f2Dt = dt.get(f2.getKey());
    f1Dt.setProperty("value", "baz");
    f2Dt.setProperty("value", "boz");
    dt.put(f1Dt);
    dt.put(f2Dt);
    dt.commit();

    GetResponse response = wds.get(f1.getKey());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
    assertEquals("baz", response.getEntity().getProperty("value"));
    
    response = wds.get(f2.getKey());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
    assertEquals("boz", response.getEntity().getProperty("value"));    
    
    assertDtCompletelyRemoved(dt);
  }  
  
  /**
   * Test that updating a named entity within a transaction is executed properly.
   */
  public void testUpdateNamedEntity() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtNamedEntity("foo", "bar", "dt__baz");
    Entity f2 = createAndSaveDtNamedEntity("fab", "bab", "dt__baz");

    dt.begin();
    Entity f1Dt = dt.get(f1.getKey());
    Entity f2Dt = dt.get(f2.getKey());
    f1Dt.setProperty("value", "baz");
    f2Dt.setProperty("value", "boz");
    dt.put(f1Dt);
    dt.put(f2Dt);
    dt.commit();

    GetResponse response = wds.get(f1.getKey());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
    assertEquals("baz", response.getEntity().getProperty("value"));
    
    response = wds.get(f2.getKey());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
    assertEquals("boz", response.getEntity().getProperty("value"));    
    
    assertDtCompletelyRemoved(dt);    
  }
  
  /**
   * Test that deleting a regular entity within a transaction is executed properly.
   */
  public void testDeleteEntity() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createAndSaveDtEntity("fab", "bab");

    dt.begin();
    dt.delete(f1.getKey());
    dt.delete(f2.getKey());
    dt.commit();

    try {
      ds.get(f1.getKey());
      fail("Entity should have been deleted.");
    } catch (EntityNotFoundException e) { /* Good */ }

    try {
      ds.get(f2.getKey());
      fail("Entity should have been deleted.");
    } catch (EntityNotFoundException e) { /* Good */ }    
    
    assertDtCompletelyRemoved(dt);
  }
  
  /**
   * Test that deleting a named entity within a transaction is executed properly.
   */
  public void testDeleteNamedEntity() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtNamedEntity("foo", "bar", "dt__baz");
    Entity f2 = createAndSaveDtNamedEntity("fab", "bab", "dt__baz");

    dt.begin();
    dt.delete(f1.getKey());
    dt.delete(f2.getKey());
    dt.commit();

    try {
      ds.get(f1.getKey());
      fail("Entity should have been deleted.");
    } catch (EntityNotFoundException e) { /* Good */ }

    try {
      ds.get(f2.getKey());
      fail("Entity should have been deleted.");
    } catch (EntityNotFoundException e) { /* Good */ }
    
    GetResponse response = wds.get(f1.getKey());
    assertNull(response.getEntity());
    assertNotNull(response.getPureMetaData());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());

    response = wds.get(f2.getKey());
    assertNull(response.getEntity());
    assertNotNull(response.getPureMetaData());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());    
    
    assertDtCompletelyRemoved(dt);
  }
  
  public void testReadConsistent() throws Exception {
    DistributedTransaction dt1 = new DistributedTransaction(ds);
    DistributedTransaction dt2 = new DistributedTransaction(ds, true);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt1.begin();
    Entity f1Dt1Copy = dt1.get(f1.getKey());
    f1Dt1Copy.setProperty("value", "baz");
    dt1.put(f1Dt1Copy);
    dt1.initialize();
    dt1.acquireWriteLocks();
    dt1.checkReadVersions();
    
    // Since this DT was told to "read consistent", pending DTs will be rolled forward during get.
    dt2.begin();
    dt2.get(f1.getKey());
    
    GetResponse response = wds.get(f1.getKey());
    assertEquals(null, response.getWriteLock());
    assertEquals(dt1.getKey(), response.getVersion());
    assertEquals("baz", response.getEntity().getProperty("value"));
    
    CacheRecord record = dt2.getCache().get(f1.getKey());
    assertCacheRecordProps(record, null, false, false, true, false, dt1.getKey());
  }
  
  public void testReadConsistent_writeLockDtNonExistant() throws Exception {
    Key fakeDtKey = KeyFactory.createKey(DistributedTransactionEntity.KIND, "fake");
    DistributedTransaction dt2 = new DistributedTransaction(ds, true);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    f1.setProperty(WrappedDatastoreService.WRITE_LOCK_PROPERTY, fakeDtKey);
    ds.put(f1);
    
    // The blocking DT entity doesn't exist, so the read should move on and ignore this write lock.
    dt2.begin();
    dt2.get(f1.getKey());
    
    GetResponse response = wds.get(f1.getKey());
    assertEquals(fakeDtKey, response.getWriteLock());
    assertEquals(dummyVersion, response.getVersion());
    
    CacheRecord record = dt2.getCache().get(f1.getKey());
    assertCacheRecordProps(record, null, false, false, true, false, dummyVersion);    
  }
  
  /**
   * Tests that a basic commit with a mix all operation types occurs correctly.
   */
  public void testCommitMixedOperations() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createAndSaveDtNamedEntity("fab", "bab", "dt__baz");
    Entity f3 = createAndSaveDtNamedEntity("fob", "bob", "dt__boz");

    dt.begin();
    Entity f1Dt = dt.get(f1.getKey());
    Entity f2Dt = dt.get(f2.getKey());
    Entity f3Dt = dt.get(f3.getKey());
    f1Dt.setProperty("value", "baz");
    f2Dt.setProperty("value", "boz");
    dt.put(f1Dt);
    dt.put(f2Dt);
    dt.delete(f3Dt.getKey());
    dt.commit();

    GetResponse response = wds.get(f1.getKey());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
    assertEquals("baz", response.getEntity().getProperty("value"));
    
    response = wds.get(f2.getKey());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
    assertEquals("boz", response.getEntity().getProperty("value"));    
    
    response = wds.get(f3.getKey());
    assertNull(response.getEntity());
    assertNotNull(response.getPureMetaData());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());    
    
    assertDtCompletelyRemoved(dt);
  }    
  
  public void testMultipleEntityGroups() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity foo = createAndSaveDtEntity("foo", "a");
    Entity bar = createAndSaveDtEntity("bar", "a");

    dt.begin();
    Entity fooDt = dt.get(foo.getKey());
    Entity barDt = dt.get(bar.getKey());
    fooDt.setProperty("value", "b");
    barDt.setProperty("value", "b");
    dt.put(fooDt);
    dt.put(barDt);
    dt.commit();

    GetResponse response = wds.get(foo.getKey());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
    assertEquals("b", response.getEntity().getProperty("value"));
    
    response = wds.get(bar.getKey());
    assertNull(response.getWriteLock());
    assertEquals(dt.getKey(), response.getVersion());
    assertEquals("b", response.getEntity().getProperty("value"));
  }  
  
  ///// Record cache /////
  
  public void testCacheGet() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createAndSaveDtEntity("foo", "bar");
    Key nonExistantKey = KeyFactory.createKey("fake", 1L);
    Map<Key, CacheRecord> cache = dt.getCache();
    
    dt.begin();
    Entity f1Copy1 = dt.get(f1.getKey());
    Entity f1Copy2 = dt.get(f1.getKey());
    
    CacheRecord record = cache.get(f1.getKey());
    assertCacheRecordProps(record, f1.getProperties(), false, false, true, false, dummyVersion);
    assertEquals(f1Copy1, f1Copy2);
    
    try {
      dt.get(nonExistantKey);
      fail("The key should not exist in the datastore");
    } catch (EntityNotFoundException e) { /* Good */ }

    try {
      dt.get(nonExistantKey);
      fail("The key should not exist in the datastore");
    } catch (EntityNotFoundException e) { /* Good */ }    
    
    record = cache.get(nonExistantKey);
    assertCacheRecordProps(record, null, false, false, true, false, null);
    
    dt.put(f2);
    try {
      dt.get(f2.getKey());
      fail("Trying to read after a blind put should fail");
    } catch (RuntimeException e) { /* Good */ }
  }
  
  public void testCachePut() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createDtEntity("foo", "bar");
    Map<Key, CacheRecord> cache = dt.getCache();
    
    dt.begin();
    Entity f1DtCopy = dt.get(f1.getKey());
    dt.put(f1);
    Key k1 = dt.put(f2);
    
    CacheRecord record = cache.get(f1.getKey());
    assertCacheRecordProps(record, f1.getProperties(), true, false, true, false, dummyVersion);
    
    f1DtCopy.setProperty("value", "baz");
    dt.put(f1DtCopy);    
    
    record = cache.get(f1.getKey());
    assertCacheRecordProps(record, f1DtCopy.getProperties(), true, false, true, false, 
        dummyVersion);    
        
    record = cache.get(k1);
    assertCacheRecordProps(record, f2.getProperties(), true, true, false, false, null);
    
    // Delete an entity to replace its cache entry.
    dt.delete(f1.getKey());
    record = cache.get(f1.getKey());
    assertCacheRecordProps(record, null, false, false, true, true, dummyVersion);
    
    // Now put it again to replace the delete entry.
    dt.put(f1DtCopy);
    record = cache.get(f1.getKey());
    assertCacheRecordProps(record, f1DtCopy.getProperties(), true, false, true, false, 
        dummyVersion);        
  }
  
  public void testCacheDelete() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createAndSaveDtEntity("foo", "bar");
    Map<Key, CacheRecord> cache = dt.getCache();
    
    dt.begin();
    dt.get(f1.getKey());
    dt.delete(f1.getKey());
    dt.delete(f2.getKey());
    
    CacheRecord record = cache.get(f1.getKey());
    assertCacheRecordProps(record, null, false, false, true, true, dummyVersion);
       
    record = cache.get(f2.getKey());
    assertCacheRecordProps(record, null, false, false, false, true, null);
    
    // Put an entity to replace its cache entry.
    dt.put(f1);
    record = cache.get(f1.getKey());
    assertCacheRecordProps(record, f1.getProperties(), true, false, true, false, dummyVersion);
    
    // Now delete it again to restore the delete entry.
    dt.delete(f1.getKey());
    record = cache.get(f1.getKey());
    assertCacheRecordProps(record, null, false, false, true, true, dummyVersion);
  }
  
  ///// State progression tests /////
  
  /**
   * Test that beginning a transaction merely puts it in the datastore to allocate a DT key.  
   */
  public void testBegin() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    dt.begin();

    assertTrue(dt.getCache().isEmpty());
    
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    assertEquals(State.INIT_0, dtCopy.getState());
    assertTrue(dtCopy.getGets().isEmpty());
    assertTrue(dtCopy.getPuts().isEmpty());
  }
  
  /**
   * 
   */
  public void testBeginAndRecord() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createDtEntity("foo", "bar");
    Entity f2 = createAndSaveDtEntity("foo", "bar");
    Entity f3 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Key k1 = dt.put(f1);
    Entity f2Dt = dt.get(f2.getKey());
    f2Dt.setProperty("value", "baz");
    dt.put(f2Dt);
    dt.delete(f3.getKey());
    
    Map<Key, CacheRecord> cache = dt.getCache();
    
    CacheRecord record = cache.get(k1);
    assertCacheRecordProps(record, f1.getProperties(), true, true, false, false, null);
    
    record = cache.get(f2.getKey());
    assertCacheRecordProps(record, null, true, false, true, false, dummyVersion);
    
    record = cache.get(f3.getKey());
    assertCacheRecordProps(record, null, false, false, false, true, null);
    
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    assertEquals(State.INIT_0, dtCopy.getState());
    assertTrue(dtCopy.getGets().isEmpty());
    assertTrue(dtCopy.getPuts().isEmpty());
    
    try {
      ds.get(k1);
    } catch (EntityNotFoundException e) { /* Good */ }
    
    f2 = ds.get(f2.getKey());
    assertEquals("bar", f2.getProperty("value"));
    
    ds.get(f3.getKey());
  }
  
  /**
   * Tests that a transaction is initialized correctly.  This means that it
   * should contain a transitive closure of all work that must be performed.
   */
  public void testInitialize() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");

    dt.begin();
    f1 = dt.get(f1.getKey());
    f1.setProperty("value", "baz");
    dt.put(f1);
    dt.initialize();

    Map<Key, CacheRecord> cache = dt.getCache();
    
    CacheRecord record = cache.get(f1.getKey());
    assertCacheRecordProps(record, null, true, false, true, false, dummyVersion);
    
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getEntity().getKey());
    assertEquals(State.READY_1, dtCopy.getState());
    assertEquals(f1.getKey(), dtCopy.getGets().get(0).getKey());
    assertEquals(f1.getKey(), dtCopy.getPuts().get(0).getKey());
    
    GetResponse response = wds.get(f1.getKey());
    assertEquals(null, response.getWriteLock());
    assertEquals(dummyVersion, response.getVersion());
    assertEquals("bar", response.getEntity().getProperty("value"));    
  }
  
  /**
   * Test that gets and puts are sorted correctly and consistently.
   */
  public void testSortGetsAndPuts() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createAndSaveDtEntity("foo", "baz");
    Entity newFoo1 = createDtEntity("foo", "bar");
    Entity newFoo2 = createDtEntity("foo", "baz");
    Entity newFoo3 = createDtEntityWithParent("foo", "bar", f1.getKey());

    dt.begin();
    dt.get(f2.getKey());
    dt.get(f1.getKey());
    dt.put(f2);
    dt.put(f1);
    Key k1 = dt.put(newFoo1);
    Key k2 = dt.put(newFoo2);
    Key k3 = dt.put(newFoo3);
    dt.initialize();

    assertEquals(f1.getKey(), dt.getGets().get(0).getKey());
    assertEquals(f2.getKey(), dt.getGets().get(1).getKey());
    assertEquals(f1.getKey(), dt.getPuts().get(0).getKey());
    assertEquals(k3, dt.getPuts().get(1).getKey());
    assertEquals(f2.getKey(), dt.getPuts().get(2).getKey());
    assertEquals(k1, dt.getPuts().get(3).getKey());
    assertEquals(k2, dt.getPuts().get(4).getKey());
  }  
  
  public void testAcquireWriteLocks() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    f1 = dt.get(f1.getKey());
    f1.setProperty("value", "baz");
    dt.put(f1);
    dt.initialize();
    dt.acquireWriteLocks();
    
    Map<Key, CacheRecord> cache = dt.getCache();    
    
    CacheRecord record = cache.get(f1.getKey());
    assertCacheRecordProps(record, null, true, false, true, false, dummyVersion);
    
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getEntity().getKey());
    assertEquals(State.LOCKED_2, dtCopy.getState());
    assertEquals(f1.getKey(), dtCopy.getGets().get(0).getKey());
    assertEquals(f1.getKey(), dtCopy.getPuts().get(0).getKey());
    
    GetResponse response = wds.get(f1.getKey());
    assertEquals(dtCopy.getEntity().getKey(), response.getWriteLock());
    assertEquals(dummyVersion, response.getVersion());
    assertEquals("bar", response.getEntity().getProperty("value"));
  }
  
  public void testCheckReadVersions() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    f1 = dt.get(f1.getKey());
    f1.setProperty("value", "baz");
    dt.put(f1);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    
    Map<Key, CacheRecord> cache = dt.getCache();    
    
    CacheRecord record = cache.get(f1.getKey());
    assertCacheRecordProps(record, null, true, false, true, false, dummyVersion);    
    
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getEntity().getKey());
    assertEquals(State.CHECKED_3, dtCopy.getState());
    assertEquals(f1.getKey(), dtCopy.getGets().get(0).getKey());
    assertEquals(f1.getKey(), dtCopy.getPuts().get(0).getKey());
    
    GetResponse response = wds.get(f1.getKey());
    assertEquals(dtCopy.getEntity().getKey(), response.getWriteLock());
    assertEquals(dummyVersion, response.getVersion());
    assertEquals("bar", response.getEntity().getProperty("value"));    
  }
  
  public void testAbort_readVersionChanged() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    f1 = dt.get(f1.getKey());
    f1.setProperty("value", "baz");
    dt.put(f1);
    dt.initialize();
    dt.acquireWriteLocks();
    
    // Force an abort.
    Key dtKey = KeyFactory.createKey(DistributedTransactionEntity.KIND, 10L);
    Entity f1Copy = ds.get(f1.getKey());
    f1Copy.setProperty(WrappedDatastoreService.VERSION_PROPERTY, dtKey);
    ds.put(f1Copy);
    
    try {
      dt.checkReadVersions();
      fail("This should have aborted on a read version change");
    } catch (ReadVersionChangedException e) {
      dt.abort();  
    }
    
    Map<Key, CacheRecord> cache = dt.getCache();    
    
    CacheRecord record = cache.get(f1.getKey());
    assertCacheRecordProps(record, null, true, false, true, false, dummyVersion);
    
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getEntity().getKey());
    assertEquals(State.ABORTING_3, dtCopy.getState());
    assertEquals(f1.getKey(), dtCopy.getGets().get(0).getKey());
    assertEquals(f1.getKey(), dtCopy.getPuts().get(0).getKey());
    
    GetResponse response = wds.get(f1.getKey());
    assertEquals(dt.getKey(), response.getWriteLock());
    assertEquals(dtKey, response.getVersion());
    assertEquals("bar", response.getEntity().getProperty("value"));     
  }  
  
  public void testAbort_writeLockNotHeld() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    f1 = dt.get(f1.getKey());
    dt.initialize();
    dt.acquireWriteLocks();
    
    // Force an abort.
    Key dtKey = KeyFactory.createKey(DistributedTransactionEntity.KIND, 10L);
    Entity f1Copy = ds.get(f1.getKey());
    f1Copy.setProperty(WrappedDatastoreService.WRITE_LOCK_PROPERTY, dtKey);
    ds.put(f1Copy);
    
    try {
      dt.checkReadVersions();
      fail("This should have aborted on a read version change");
    } catch (WriteLockHeldException e) {
      dt.abort();  
    }
    
    Map<Key, CacheRecord> cache = dt.getCache();    
    
    CacheRecord record = cache.get(f1.getKey());
    assertCacheRecordProps(record, null, false, false, true, false, dummyVersion);
    
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getEntity().getKey());
    assertEquals(State.ABORTING_3, dtCopy.getState());
    assertEquals(f1.getKey(), dtCopy.getGets().get(0).getKey());
    
    GetResponse response = wds.get(f1.getKey());
    assertEquals(dtKey, response.getWriteLock());
    assertEquals(dummyVersion, response.getVersion());
    assertEquals("bar", response.getEntity().getProperty("value"));    
  }
  
  public void testCopy() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    f1 = dt.get(f1.getKey());
    f1.setProperty("value", "baz");
    dt.put(f1);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    dt.copy();
    
    Map<Key, CacheRecord> cache = dt.getCache();    
    
    CacheRecord record = cache.get(f1.getKey());
    assertCacheRecordProps(record, null, true, false, true, false, dummyVersion); 
    
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getEntity().getKey());
    assertEquals(State.DONE_4, dtCopy.getState());
    assertEquals(f1.getKey(), dtCopy.getGets().get(0).getKey());
    assertEquals(f1.getKey(), dtCopy.getPuts().get(0).getKey());
    
    GetResponse response = wds.get(f1.getKey());
    assertEquals(null, response.getWriteLock());
    assertEquals(dt.getEntity().getKey(), response.getVersion());
    assertEquals("baz", response.getEntity().getProperty("value"));    
  }
  
  public void testCleanup() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    f1 = dt.get(f1.getKey());
    f1.setProperty("value", "baz");
    dt.put(f1);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    dt.abort();
    dt.cleanup();
    
    Map<Key, CacheRecord> cache = dt.getCache();    
    
    CacheRecord record = cache.get(f1.getKey());
    assertCacheRecordProps(record, null, true, false, true, false, dummyVersion);
    
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getEntity().getKey());
    assertEquals(State.ABORTED_4, dtCopy.getState());
    assertEquals(f1.getKey(), dtCopy.getGets().get(0).getKey());
    assertEquals(f1.getKey(), dtCopy.getPuts().get(0).getKey());
    
    GetResponse response = wds.get(f1.getKey());
    assertEquals(null, response.getWriteLock());
    assertEquals(dummyVersion, response.getVersion());
    assertEquals("bar", response.getEntity().getProperty("value"));    
    
    assertShadowsCompletelyRemoved(dt);
  }
  
  public void testComplete_fromDone() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    f1 = dt.get(f1.getKey());
    f1.setProperty("value", "baz");
    dt.put(f1);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    dt.copy();
    dt.complete();
    
    Map<Key, CacheRecord> cache = dt.getCache();    
    
    CacheRecord record = cache.get(f1.getKey());
    assertCacheRecordProps(record, null, true, false, true, false, dummyVersion);      
    
    GetResponse response = wds.get(f1.getKey());
    assertEquals(null, response.getWriteLock());
    assertEquals(dt.getEntity().getKey(), response.getVersion());
    assertEquals("baz", response.getEntity().getProperty("value"));
    
    assertDtCompletelyRemoved(dt);
  }
  
  public void testComplete_fromAbort() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    f1 = dt.get(f1.getKey());
    f1.setProperty("value", "baz");
    dt.put(f1);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    dt.abort();
    dt.cleanup();
    dt.complete();
    
    Map<Key, CacheRecord> cache = dt.getCache();    
    
    CacheRecord record = cache.get(f1.getKey());
    assertCacheRecordProps(record, null, true, false, true, false, dummyVersion);
    
    GetResponse response = wds.get(f1.getKey());
    assertEquals(null, response.getWriteLock());
    assertEquals(dummyVersion, response.getVersion());
    assertEquals("bar", response.getEntity().getProperty("value"));    
    
    assertDtCompletelyRemoved(dt);    
  }
  
  ///// Entity group tests /////
  
  public void testSpansSingleEntityGroup() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createDtEntityWithParent("foo", "bar", f1.getKey());
    Entity f3 = createAndSaveDtEntity("fab", "bab");
    
    dt.begin();
    dt.get(f1.getKey());
    assertTrue(dt.spansSingleEntityGroup());
    dt.put(f2);
    assertTrue(dt.spansSingleEntityGroup());
    dt.delete(f3.getKey());
    assertFalse(dt.spansSingleEntityGroup());
  }
  
  ///// Granular shadow copy tests /////
  
  /**
   * Tests to make sure generating shadow copies, even if executed multiple times, creates the
   * same result.  Ancestor queries are used to ensure that the query is always consistent, even
   * on production.
   */
  public void testGenerateShadowCopies_ensureIdempotent() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createDtEntityWithParent("foo", "bar", f1.getKey());
    
    dt.begin();
    dt.put(f1);
    dt.put(f2);
    
    dt.initialize();
    Transaction txn = ds.beginTransaction();
    assertEquals(2, ds.prepare(txn, new Query(ShadowEntity.KIND, f1.getKey())).countEntities());
    txn.commit();
    
    dt.initialize();
    txn = ds.beginTransaction();
    assertEquals(2, ds.prepare(txn, new Query(ShadowEntity.KIND, f1.getKey())).countEntities());
    txn.commit();
  }
  
  ///// Granular acquire write lock tests /////
  
  public void testAcquireWriteLocks_simulateTwoThreads() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createAndSaveDtEntity("foo", "baz");
    
    dt.begin();
    dt.put(f1);
    dt.put(f2);
    dt.initialize();
    
    for (PutRequest putRequest : dt.getPuts()) {
      assertTrue(dt.acquireWriteLock(putRequest));
      assertFalse(dt.acquireWriteLock(putRequest));
    }    
  }
  
  public void testAcquireWriteLocks_ensureNoLockingIfBehind() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createAndSaveDtEntity("foo", "baz");
    
    dt.begin();
    dt.put(f1);
    dt.put(f2);
    dt.initialize();
    
    for (PutRequest putRequest : dt.getPuts()) {
      // Simulate another thread getting ahead (copying or aborting) by deleting the shadow.
      ds.delete(putRequest.getShadowKey());
      assertFalse(dt.acquireWriteLock(putRequest));
    }       
  }
  
  public void testAcquireWriteLocks_rollFowardBlockingLock() throws Exception {
    DistributedTransaction dt1 = new DistributedTransaction(ds);
    DistributedTransaction dt2 = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createAndSaveDtEntity("foz", "boz");

    dt1.begin();
    Entity f1Dt = dt1.get(f1.getKey());
    f1Dt.setProperty("value", "baz");
    dt1.put(f1Dt);
    dt1.initialize();
    dt1.acquireWriteLocks();

    GetResponse response = wds.get(f1.getKey());
    assertEquals(dt1.getKey(), response.getWriteLock());
    assertEquals(dummyVersion, response.getVersion());
    assertEquals("bar", response.getEntity().getProperty("value"));
    
    dt2.begin();
    f1Dt = dt2.get(f1.getKey());
    Entity f2Dt = dt2.get(f2.getKey());
    f1Dt.setProperty("value", "boz");
    f2Dt.setProperty("value", "bob");
    dt2.put(f1Dt);
    dt2.put(f2Dt);
    dt2.initialize();
    
    List<PutRequest> putRequests = dt2.getPuts();
    assertEquals(2, putRequests.size());
    
    PutRequest putRequest = putRequests.get(0);
    dt2.acquireWriteLock(putRequest);
    
    response = wds.get(f1.getKey());
    assertEquals(dt2.getKey(), response.getWriteLock());
    assertEquals(dt1.getKey(), response.getVersion());
    assertEquals("baz", response.getEntity().getProperty("value"));
    
    putRequest = putRequests.get(1);
    dt2.acquireWriteLock(putRequest);
    
    response = wds.get(f2.getKey());
    assertEquals(dt2.getKey(), response.getWriteLock());
    assertEquals(dummyVersion, response.getVersion());
    assertEquals("boz", response.getEntity().getProperty("value"));    
  }
  
  public void testAcquireWriteLocks_ignoreNonExistantWriteLock() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    dt.put(f1);
    dt.initialize();
    
    // Set the write lock.
    Key fakeDtKey = KeyFactory.createKey(DistributedTransactionEntity.KIND, 10L);
    f1.setProperty(WrappedDatastoreService.WRITE_LOCK_PROPERTY, fakeDtKey);
    ds.put(f1);
    
    for (PutRequest putRequest : dt.getPuts()) {
      assertTrue(dt.acquireWriteLock(putRequest));
    }      
  }
  
  ///// Granular check read version tests /////
  
  public void testCheckReadVersions_versionAndWriteLockNull() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createDtNamedEntity("foo", "bar", "dt__baz");
   
    dt.begin();
    try {
      dt.get(f1.getKey());
      fail("This entity should not exist in the datastore yet.");
    } catch (EntityNotFoundException e) { /* Good */ }
    dt.initialize();
    dt.acquireWriteLocks();

    // Change the version.
    Key dtKey = KeyFactory.createKey(DistributedTransactionEntity.KIND, 10L);    
    GetResponse response = wds.get(f1.getKey());
    Entity pureMetaData = response.getPureMetaData();
    pureMetaData.setProperty(WrappedDatastoreService.VERSION_PROPERTY, dtKey);
    ds.put(Lists.newArrayList(f1, pureMetaData));
    
    try {
      dt.checkReadVersions();
    } catch (ReadVersionChangedException e) { /* Good */ }    
  }
  
  public void testCheckReadVersions_readVersionChanged() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    dt.get(f1.getKey());
    dt.initialize();
    dt.acquireWriteLocks();
    
    // Change the version.
    Key dtKey = KeyFactory.createKey(DistributedTransactionEntity.KIND, 10L);
    f1.setProperty(WrappedDatastoreService.VERSION_PROPERTY, dtKey);
    ds.put(f1);
    
    try {
      dt.checkReadVersions();
    } catch (ReadVersionChangedException e) { /* Good */ }
  }
  
  public void testCheckReadVersions_writeLockHeld() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    dt.get(f1.getKey());
    dt.initialize();
    dt.acquireWriteLocks();
    
    // Change the version.
    Key dtKey = KeyFactory.createKey(DistributedTransactionEntity.KIND, 10L);
    f1.setProperty(WrappedDatastoreService.WRITE_LOCK_PROPERTY, dtKey);
    ds.put(f1);

    try {
      dt.checkReadVersions();
    } catch (WriteLockHeldException e) { /* Good */ }    
  }
  
  ///// Granular copy logic tests /////
  
  /**
   * Test that two threads rolling forward the copying shadow entities do not repeat work.
   */
  public void testCopyShadowEntity_simulateTwoThreads() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createAndSaveDtEntity("foo", "baz");
    
    dt.begin();
    dt.put(f1);
    dt.put(f2);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    
    for (PutRequest putRequest : dt.getPuts()) {
      Transaction txn = ds.beginTransaction();
      assertTrue(dt.copyShadowEntity(txn, putRequest));
      txn.commit();
      
      txn = ds.beginTransaction();
      assertFalse(dt.copyShadowEntity(txn, putRequest));
      txn.commit();
    }
  }
  
  /**
   * Test the logic that groups shadow copies by entity group.
   */
  public void testCopyShadowEntities_groupByEntityGroup() throws Exception {
    DatastoreService mockDs = EasyMock.createMock(DatastoreService.class);
    Transaction mockTxn = EasyMock.createMock(Transaction.class);
    DistributedTransaction dt = new DistributedTransaction(mockDs);
    ds.put(dt.getDtEntity().getEntity());
    Key dtKey = dt.getKey();
    
    // Setup two entity groups with a mix of puts and deletes.
    Entity f1Grp1 = createAndSaveDtEntity("foo", "bar");
    Entity f2Grp1 = createDtEntityWithParent("foo", "bar", f1Grp1.getKey());
    ds.put(f2Grp1);
    Entity f3Grp2 = createAndSaveDtEntity("baz", "bar");
    Entity f4Grp2 = createDtEntityWithParent("baz", "bar", f3Grp2.getKey());
    ds.put(f4Grp2);
    ShadowEntity f1Shadow = ShadowEntity.createAndSavePutShadow(ds, dtKey, f1Grp1);
    ShadowEntity f2Shadow = ShadowEntity.createAndSavePutShadow(ds, dtKey, f2Grp1);
    ShadowEntity f3Shadow = ShadowEntity.createAndSaveDeleteShadow(ds, dtKey, f3Grp2.getKey());
    ShadowEntity f4Shadow = ShadowEntity.createAndSavePutShadow(ds, dtKey, f4Grp2);
    PutRequest p1 = new PutRequest(f1Grp1.getKey(), f1Shadow.getKey(), false);
    PutRequest p2 = new PutRequest(f2Grp1.getKey(), f2Shadow.getKey(), true);
    PutRequest p3 = new PutRequest(f3Grp2.getKey(), f3Shadow.getKey(), false);
    PutRequest p4 = new PutRequest(f4Grp2.getKey(), f4Shadow.getKey(), true);
    
    // Setup expectations.
    dt.getDtEntity().setPuts(Lists.newArrayList(p1, p2, p3, p4));
    
    // There are exactly 2 entity groups, so there should be 2 commits.
    EasyMock.expect(mockDs.beginTransaction()).andReturn(mockTxn).times(2);
    EasyMock.expect(mockTxn.isActive()).andReturn(true).times(2);
    mockTxn.commit();
    EasyMock.expectLastCall().times(2);
    
    // Copy the first shadow.
    EasyMock.expect(mockDs.get(mockTxn, p1.getShadowKey())).andReturn(f1Shadow.getEntity());
    EasyMock.expect(mockDs.put(mockTxn, Lists.newArrayList(f1Shadow.getShadowed()))).andReturn(null);
    mockDs.delete(mockTxn, p1.getShadowKey());

    // Copy the second shadow.
    EasyMock.expect(mockDs.get(mockTxn, p2.getShadowKey())).andReturn(f2Shadow.getEntity());
    EasyMock.expect(mockDs.put(mockTxn, Lists.newArrayList(f2Shadow.getShadowed()))).andReturn(null);
    mockDs.delete(mockTxn, p2.getShadowKey());
    
    // Copy the third shadow.    
    EasyMock.expect(mockDs.get(mockTxn, p3.getShadowKey())).andReturn(f3Shadow.getEntity());
    mockDs.delete(mockTxn, p3.getKey());
    mockDs.delete(mockTxn, p3.getShadowKey());
    
    // Copy the fourth shadow.    
    EasyMock.expect(mockDs.get(mockTxn, p4.getShadowKey())).andReturn(f4Shadow.getEntity());
    EasyMock.expect(mockDs.put(mockTxn, Lists.newArrayList(f4Shadow.getShadowed()))).andReturn(null);
    mockDs.delete(mockTxn, p4.getShadowKey());
    
    EasyMock.expect(mockTxn.isActive()).andReturn(false);
    
    EasyMock.replay(mockDs, mockTxn);
    
    dt.copyShadowEntities();
    
    EasyMock.verify(mockDs, mockTxn);
  }
  
  ///// Granular cleanup logic tests /////
  
  public void testCleanupPutRequest_simulateTwoThreads() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    Entity f2 = createAndSaveDtEntity("foo", "baz");
    
    dt.begin();
    dt.put(f1);
    dt.put(f2);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    
    for (PutRequest putRequest : dt.getPuts()) {
      Transaction txn = ds.beginTransaction();
      assertTrue(dt.cleanupPutRequest(txn, putRequest));
      txn.commit();
      
      txn = ds.beginTransaction();
      assertFalse(dt.cleanupPutRequest(txn, putRequest));
      txn.commit();
    }    
  }
  
  public void testCleanupPutRequests_groupByEntityGroup() throws Exception {
    DatastoreService mockDs = EasyMock.createMock(DatastoreService.class);
    Transaction mockTxn = EasyMock.createMock(Transaction.class);
    DistributedTransaction dt = new DistributedTransaction(mockDs);
    ds.put(dt.getDtEntity().getEntity());
    Key dtKey = dt.getKey();
    
    // Setup two entity groups with a mix of puts and deletes.
    Entity f1Grp1 = createAndSaveDtEntity("foo", "bar");
    Entity f2Grp1 = createDtEntityWithParent("foo", "bar", f1Grp1.getKey());
    ds.put(f2Grp1);
    Entity f3Grp2 = createAndSaveDtEntity("baz", "bar");
    Entity f4Grp2 = createDtEntityWithParent("baz", "bar", f3Grp2.getKey());
    ds.put(f4Grp2);
    ShadowEntity f1Shadow = ShadowEntity.createAndSavePutShadow(ds, dtKey, f1Grp1);
    ShadowEntity f2Shadow = ShadowEntity.createAndSavePutShadow(ds, dtKey, f2Grp1);
    ShadowEntity f3Shadow = ShadowEntity.createAndSaveDeleteShadow(ds, dtKey, f3Grp2.getKey());
    ShadowEntity f4Shadow = ShadowEntity.createAndSavePutShadow(ds, dtKey, f4Grp2);
    PutRequest p1 = new PutRequest(f1Grp1.getKey(), f1Shadow.getKey(), false);
    PutRequest p2 = new PutRequest(f2Grp1.getKey(), f2Shadow.getKey(), true);
    PutRequest p3 = new PutRequest(f3Grp2.getKey(), f3Shadow.getKey(), false);
    PutRequest p4 = new PutRequest(f4Grp2.getKey(), f4Shadow.getKey(), true);
    
    // Lock all of the entities.
    f1Grp1.setProperty(WrappedDatastoreService.WRITE_LOCK_PROPERTY, dtKey);
    f2Grp1.setProperty(WrappedDatastoreService.WRITE_LOCK_PROPERTY, dtKey);
    f3Grp2.setProperty(WrappedDatastoreService.WRITE_LOCK_PROPERTY, dtKey);
    f4Grp2.setProperty(WrappedDatastoreService.WRITE_LOCK_PROPERTY, dtKey);
    
    // Setup expectations.
    dt.getDtEntity().setPuts(Lists.newArrayList(p1, p2, p3, p4));
    
    // There are exactly 2 entity groups, so there should be 2 commits.
    EasyMock.expect(mockDs.beginTransaction()).andReturn(mockTxn).times(2);
    EasyMock.expect(mockTxn.isActive()).andReturn(true).times(2);
    mockTxn.commit();
    EasyMock.expectLastCall().times(2);
    
    // Cleanup the first shadow.
    EasyMock.expect(mockDs.get(mockTxn, p1.getShadowKey())).andReturn(f1Shadow.getEntity());
    EasyMock.expect(mockDs.get(mockTxn, p1.getKey())).andReturn(f1Grp1);
    EasyMock.expect(mockDs.put(mockTxn, f1Grp1)).andReturn(null);
    mockDs.delete(mockTxn, p1.getShadowKey());

    // Cleanup the second shadow.
    EasyMock.expect(mockDs.get(mockTxn, p2.getShadowKey())).andReturn(f2Shadow.getEntity());
    EasyMock.expect(mockDs.get(mockTxn, p2.getKey())).andReturn(f2Grp1);
    EasyMock.expect(mockDs.put(mockTxn, f2Grp1)).andReturn(null);
    mockDs.delete(mockTxn, p2.getShadowKey());
    
    // Cleanup the third shadow.    
    EasyMock.expect(mockDs.get(mockTxn, p3.getShadowKey())).andReturn(f3Shadow.getEntity());
    EasyMock.expect(mockDs.get(mockTxn, p3.getKey())).andReturn(f3Grp2);
    EasyMock.expect(mockDs.put(mockTxn, f3Grp2)).andReturn(null);
    mockDs.delete(mockTxn, p3.getShadowKey());
    
    // Cleanup the fourth shadow.    
    EasyMock.expect(mockDs.get(mockTxn, p4.getShadowKey())).andReturn(f4Shadow.getEntity());
    EasyMock.expect(mockDs.get(mockTxn, p4.getKey())).andReturn(f4Grp2);
    EasyMock.expect(mockDs.put(mockTxn, f4Grp2)).andReturn(null);
    mockDs.delete(mockTxn, p4.getShadowKey());
    
    EasyMock.expect(mockTxn.isActive()).andReturn(false);
    
    EasyMock.replay(mockDs, mockTxn);
    
    dt.cleanupPutRequests();
    
    EasyMock.verify(mockDs, mockTxn);    
  }
  
  ///// Granular acquire lock state transition tests /////
  
  public void testImpossibleBackwardsStateTransition() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    
    // Force the DT to artificially backtrack.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.getDtEntity().setState(State.READY_1);
    dtCopy.save();
    
    try {
      dt.checkReadVersions();
      fail("Impossible backwards state transition should have been caught");
    } catch (RuntimeException e) { /* Good */ }
  }
  
  public void testAcquireWriteLocks_behindAtLocked() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.acquireWriteLocks();
    
    for (PutRequest putRequest : dt.getPuts()) {
      assertFalse(dt.acquireWriteLock(putRequest));
    }
    
    dt.transitionState(State.LOCKED_2);
    assertEquals(State.LOCKED_2, dt.getState());
    assertEquals(State.LOCKED_2, dtCopy.getState());
  }
  
  public void testAcquireWriteLocks_behindAtChecked() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.acquireWriteLocks();
    dtCopy.checkReadVersions();
    
    for (PutRequest putRequest : dt.getPuts()) {
      assertFalse(dt.acquireWriteLock(putRequest));
    }
    
    dt.transitionState(State.LOCKED_2);
    assertEquals(State.CHECKED_3, dt.getState());
    assertEquals(State.CHECKED_3, dtCopy.getState());    
  }
  
  public void testAcquireWriteLocks_behindAtAbort() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.acquireWriteLocks();
    dtCopy.abort();
    
    for (PutRequest putRequest : dt.getPuts()) {
      assertFalse(dt.acquireWriteLock(putRequest));
    }
    
    dt.transitionState(State.LOCKED_2);
    assertEquals(State.ABORTING_3, dt.getState());
    assertEquals(State.ABORTING_3, dtCopy.getState());      
  }
  
  public void testAcquireWriteLocks_behindAtCopied() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.acquireWriteLocks();
    dtCopy.checkReadVersions();
    dtCopy.copy();
    
    for (PutRequest putRequest : dt.getPuts()) {
      assertFalse(dt.acquireWriteLock(putRequest));
    }
    
    dt.transitionState(State.LOCKED_2);
    assertEquals(State.DONE_4, dt.getState());
    assertEquals(State.DONE_4, dtCopy.getState());      
  }
  
  public void testAcquireWriteLocks_behindAtCleanup() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.acquireWriteLocks();
    dtCopy.checkReadVersions();
    dtCopy.abort();
    dtCopy.cleanup();
    
    for (PutRequest putRequest : dt.getPuts()) {
      assertFalse(dt.acquireWriteLock(putRequest));
    }
    
    dt.transitionState(State.LOCKED_2);
    assertEquals(State.ABORTED_4, dt.getState());
    assertEquals(State.ABORTED_4, dtCopy.getState());        
  }
  
  public void testAcquireWriteLocks_behindAtCompleted() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.acquireWriteLocks();
    dtCopy.checkReadVersions();
    dtCopy.copy();
    dtCopy.complete();
    
    for (PutRequest putRequest : dt.getPuts()) {
      assertFalse(dt.acquireWriteLock(putRequest));
    }
    
    try {
      dt.transitionState(State.LOCKED_2);
      fail("Competing thread should have completed, making this DT invalid.");
    } catch (DistributedTransactionNotFoundException e) { /* Good */ } 
    
    assertDtCompletelyRemoved(dt);
  }
  
  public void testAcquireWriteLocks_behindAtCompletedViaAbort() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.acquireWriteLocks();
    dtCopy.checkReadVersions();
    dtCopy.abort();
    dtCopy.cleanup();
    dtCopy.complete();
    
    for (PutRequest putRequest : dt.getPuts()) {
      assertFalse(dt.acquireWriteLock(putRequest));
    }
    
    try {
      dt.transitionState(State.LOCKED_2);
      fail("Competing thread should have completed, making this DT invalid.");
    } catch (DistributedTransactionNotFoundException e) { /* Good */ }
    
    assertDtCompletelyRemoved(dt);
  }
 
  ///// Granular check versions state transition tests /////
  
  public void testCheckReadVersions_behindAtChecked() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.checkReadVersions();
    
    dt.checkReadVersions();
    assertEquals(State.CHECKED_3, dt.getState());
    assertEquals(State.CHECKED_3, dtCopy.getState());
  }
  
  public void testCheckReadVersions_behindAtAbort() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.acquireWriteLocks();
    dtCopy.abort();
    
    dt.checkReadVersions();
    assertEquals(State.ABORTING_3, dt.getState());
    assertEquals(State.ABORTING_3, dtCopy.getState());      
  }
  
  public void testCheckReadVersions_behindAtCopied() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.acquireWriteLocks();
    dtCopy.checkReadVersions();
    dtCopy.copy();
    
    dt.checkReadVersions();
    assertEquals(State.DONE_4, dt.getState());
    assertEquals(State.DONE_4, dtCopy.getState());      
  }
  
  public void testCheckReadVersions_behindAtCleanup() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.acquireWriteLocks();
    dtCopy.abort();
    dtCopy.cleanup();
    
    dt.checkReadVersions();
    dt.transitionState(State.CHECKED_3);
    assertEquals(State.ABORTED_4, dt.getState());
    assertEquals(State.ABORTED_4, dtCopy.getState());        
  }
  
  public void testCheckReadVersions_behindAtCompleted() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.acquireWriteLocks();
    dtCopy.checkReadVersions();
    dtCopy.copy();
    dtCopy.complete();
    
    try {
      dt.checkReadVersions();
      fail("Competing thread should have completed, making this DT invalid.");
    } catch (DistributedTransactionNotFoundException e) { /* Good */ }  
    
    assertDtCompletelyRemoved(dt);
  }
  
  public void testCheckReadVersions_behindAtCompletedViaAbort() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.acquireWriteLocks();
    dtCopy.abort();
    dtCopy.cleanup();
    dtCopy.complete();
    
    try {
      dt.checkReadVersions();
      fail("Competing thread should have completed, making this DT invalid.");
    } catch (DistributedTransactionNotFoundException e) { /* Good */ }
    
    assertDtCompletelyRemoved(dt);
  }  
  
  ///// Granular copy state transition tests /////
  
  public void testCopy_behindAtCopy() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.copy();
    
    dt.copy();
    assertEquals(State.DONE_4, dt.getState());
    assertEquals(State.DONE_4, dtCopy.getState());    
  }
  
  public void testCopy_behindAtCompleted() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.copy();
    dtCopy.complete();
    
    try {
      dt.copy();
      fail("Competing thread should have completed, making this DT invalid.");
    } catch (DistributedTransactionNotFoundException e) { /* Good */ }
    
    assertDtCompletelyRemoved(dt);
  }
  
  ///// Granular aborting state transition tests /////
  
  public void testAbort_behindAtAbort() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.abort();
    
    dt.abort();
    assertEquals(State.ABORTING_3, dt.getState());
    assertEquals(State.ABORTING_3, dtCopy.getState());       
  }
  
  public void testAbort_behindAtCleanup() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.abort();
    dtCopy.cleanup();
    
    dt.abort();
    assertEquals(State.ABORTED_4, dt.getState());
    assertEquals(State.ABORTED_4, dtCopy.getState());
  }
  
  public void testAbort_behindAtCompleted() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.abort();
    dtCopy.cleanup();
    dtCopy.complete();
    
    try {
      dt.abort();
      fail("Competing thread should have completed, making this DT invalid.");
    } catch (DistributedTransactionNotFoundException e) { /* Good */ }
    
    assertDtCompletelyRemoved(dt);
  }  
  
  ///// Granular cleanup state transition tests /////
  
  public void testCleanup_behindAtCleanup() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    dt.abort();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.cleanup();
    
    dt.cleanup();
    assertEquals(State.ABORTED_4, dt.getState());
    assertEquals(State.ABORTED_4, dtCopy.getState());    
  }
  
  public void testCleanup_behindAtComplete() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    dt.abort();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.cleanup();
    dtCopy.complete();
    
    try {
      dt.cleanup();
      fail("Competing thread should have completed, making this DT invalid.");
    } catch (DistributedTransactionNotFoundException e) { /* Good */ }
    
    assertDtCompletelyRemoved(dt);    
  }
  
  ///// Granular complete state transition tests /////
  
  public void testComplete_behindAtComplete() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.checkReadVersions();
    dt.copy();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.complete();
    
    dt.complete();
    assertDtCompletelyRemoved(dt);    
  }
  
  public void testCompleteViaAbort_behindAtCompleteViaAbort() throws Exception {
    DistributedTransaction dt = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");
    
    dt.begin();
    Entity f1Copy = dt.get(f1.getKey());
    dt.put(f1Copy);
    dt.initialize();
    dt.acquireWriteLocks();
    dt.abort();
    dt.cleanup();
    
    // Advance to the next phase.
    DistributedTransaction dtCopy = DistributedTransaction.getTxn(ds, dt.getKey());
    dtCopy.complete();
    
    dt.complete();
    assertDtCompletelyRemoved(dt);      
  }
  
  ///// Multithread tests /////
  
  /**
   * The first of our interleaved txn scenarios.  In this one, there are two
   * transactions A & B that perform the following interleaved actions:
   * - A locks: write lock is A.key(), version is previous version
   * - B reads: write lock is A.key(), version is previous version
   * - A copies: write lock is null, version is A.key()
   * - B locks: write lock is null, version is A.key() -> B Aborts
   */
  public void testInterleavedTxns1() throws Exception {
    DistributedTransaction dt1 = new DistributedTransaction(ds);
    DistributedTransaction dt2 = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");

    dt1.begin();
    Entity f1dt1 = dt1.get(f1.getKey());
    dt1.put(f1dt1);
    dt1.initialize();
    dt1.acquireWriteLocks();
    dt1.checkReadVersions();

    GetResponse response = wds.get(f1.getKey());
    assertEquals(dt1.getKey(), response.getWriteLock());
    assertEquals(dummyVersion, response.getVersion());

    dt2.begin();
    Entity f1dt2 = dt2.get(f1.getKey());
    dt2.put(f1dt2);
    dt2.initialize();
    
    // Perform copy with DT1 in order to change the version on the entity.
    dt1.copy();

    try {
      dt2.acquireWriteLocks();
      dt2.checkReadVersions();
      fail("Version should have changed and this should have aborted");
    } catch (ReadVersionChangedException e) { /* Good */ }
    
    response = wds.get(f1.getKey());
    assertEquals(dt2.getKey(), response.getWriteLock());
    assertEquals(dt1.getKey(), response.getVersion());    
  }

  /**
   * The second of our interleaved txn scenarios.  In this one there are two
   * transactions A & B that perform the following interleaved actions:
   * - A locks: write lock is A.key(), version is previous version
   * - B reads: write lock is A.key(), version is previous version
   * - B locks: A is found via write lock and is rolled forward -> B Aborts
   */
  public void testInterleavedTxns2() throws Exception {
    DistributedTransaction dt1 = new DistributedTransaction(ds);
    DistributedTransaction dt2 = new DistributedTransaction(ds);
    Entity f1 = createAndSaveDtEntity("foo", "bar");

    dt1.begin();
    Entity f1dt1 = dt1.get(f1.getKey());
    dt1.put(f1dt1);
    dt1.initialize();
    dt1.acquireWriteLocks();
    dt1.checkReadVersions();

    GetResponse response = wds.get(f1.getKey());
    assertEquals(dt1.getKey(), response.getWriteLock());
    assertEquals(dummyVersion, response.getVersion());

    dt2.begin();
    Entity f1dt2 = dt2.get(f1.getKey());
    dt2.put(f1dt2);
    dt2.initialize();

    try {
      dt2.acquireWriteLocks();
      dt2.checkReadVersions();
      fail("Version should have changed and this should have aborted");
    } catch (ReadVersionChangedException e) { /* Good */ }

    try {
      DistributedTransaction.getTxn(ds, dt1.getKey());
      fail("The first txn should be rolled forward and deleted");
    } catch (DistributedTransactionNotFoundException e) { /* Good */ }
    
    response = wds.get(f1.getKey());
    assertEquals(dt2.getKey(), response.getWriteLock());
    assertEquals(dt1.getKey(), response.getVersion());    
  }  
  
}
